package com.gec.teams.wechat.task;

import com.gec.teams.wechat.exception.TeamsException;
import com.gec.teams.wechat.pojo.entity.MessageEntity;
import com.gec.teams.wechat.pojo.entity.MessageRefEntity;
import com.gec.teams.wechat.service.MessageService;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * 功能：
 * 作者：YumTea
 * 日期：2024/6/25 16:04
 */
@Component
@Slf4j
@Scope("prototype")
public class MessageTask {
    @Autowired
    private ConnectionFactory factory;

    @Autowired
    private MessageService messageService;

    public void send(String topic, MessageEntity entity) {
        //TODO 我感觉这里不对劲，如果是一个消息想通知多个用户怎么办？
        //先将消息插入到mongoDB,然后获取消息id
        String id = messageService.insertMessage(entity);

        try (//获取rabbitmq的连接
                Connection connection = factory.newConnection();
                //信道
             Channel channel = connection.createChannel();
        ) {
            //声明当前消息使用的队列 参数2 是否要进行持久化 参数3是否是排他的
            channel.queueDeclare(topic, true, false, false, null);
            //声明 消息头数据
            HashMap map = new HashMap();
            map.put("messageId", id);
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().headers(map).build();
            channel.basicPublish("", topic, properties, entity.getMsg().getBytes());
            log.debug("消息发送成功");
        } catch (Exception e) {
            log.error("执行异常", e);
            throw new TeamsException("向MQ发送消息失败");
        }
    }

    //接收消息
    public void receive(String topic) {
        int i = 0;
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel();
        ) {
            channel.queueDeclare(topic, true, false, false, null);
            //循环查找未读消息
            while (true) {
                GetResponse response = channel.basicGet(topic, false);
                if (response != null) {
                    //先从RabbitMQ查找还没显示过的消息
                    AMQP.BasicProperties properties = response.getProps();
                    Map<String, Object> map = properties.getHeaders();
                    String messageId = map.get("messageId").toString();
                    byte[] body = response.getBody();
                    String message = new String(body);
                    log.debug("从RabbitMQ接收的消息：" + message);

                    //获取到之后，就开始插入到MongoDB数据库里面
                    MessageRefEntity entity = new MessageRefEntity();
                    entity.setMessageId(messageId);
                    entity.setReceiverId(Integer.parseInt(topic));
                    entity.setReadFlag(false);
                    entity.setLastFlag(true);
                    messageService.insertRef(entity);

                    long deliveryTag = response.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag, false);
                    i++;
                }
                else {
                    break;
                }
            }
        } catch (Exception e) {
            log.error("执行异常", e);
            throw new TeamsException("接收消息失败");
        }
    }


    public void deleteQueue(String topic){
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel();
        ) {
            channel.queueDelete(topic);
            log.debug("消息队列成功删除");
        }catch (Exception e) {
            log.error("删除队列失败", e);
            throw new TeamsException("删除队列失败");
        }
    }

    @Async
    public void receiveAsync(String topic) {
       receive(topic);
    }

    @Async
    public void deleteQueueAsync(String topic){
        deleteQueue(topic);
    }
    @Async
    public void sendAsync(String topic, MessageEntity entity) {
        send(topic, entity);
    }
}
